#pragma once
#ifndef __MY_CHANNEL__
#define __MY_CHANNEL__

#include "muduo/net/TcpConnection.h"
#include "muduo/proto/codec.h"
#include "../mqcommon/protocol.pb.h"
#include "./virtual_host.hpp"
#include "./consumer.hpp"
#include "./route.hpp"
#include "../mqcommon/thread_pool.hpp"
#include <iostream>
#include <memory>
#include <string>

/**
 * 当我们进行网络服务时，客户端与服务端之间会建立网络连接，但是在实际应用中我们发现，一个TCP连接在实际网络通信中，被使用的效率太低
 * 为了充分的使用网络通信，我们使用了更小粒度的信道来进行网络中的服务与数据传输，而信道是依赖于TCP连接的，并且一个连接可以有多个信道
 * 这些信道可以并发的处理一些网络业务
 *
 * 所以信道就是用来替代连接来为客户端提供服务的
 */

namespace NS_Channel
{
    using ProtobufCodecPtr = std::shared_ptr<ProtobufCodec>;

    using OpenChannelRequestPtr = std::shared_ptr<NS_Protocol::OpenChannelRequest>;
    using CloseChannelRequestPtr = std::shared_ptr<NS_Protocol::CloseChannelRequest>;

    using DeclareExchangeRequestPtr = std::shared_ptr<NS_Protocol::DeclareExchangeRequest>;
    using DeleteExchangeRequestPtr = std::shared_ptr<NS_Protocol::DeleteExchangeRequest>;

    using DeclareMsgQueueRequestPtr = std::shared_ptr<NS_Protocol::DeclareMsgQueueRequest>;
    using DeleteMsgQueueRequestPtr = std::shared_ptr<NS_Protocol::DeleteMsgQueueRequest>;

    using CreateBindingRequestPtr = std::shared_ptr<NS_Protocol::CreateBindingRequest>;
    using DeleteBindingRequestPtr = std::shared_ptr<NS_Protocol::DeleteBindingRequest>;

    using CreateConsumerRequestPtr = std::shared_ptr<NS_Protocol::CreateConsumerRequest>;
    using DeleteConsumerRequestPtr = std::shared_ptr<NS_Protocol::DeleteConsumerRequest>;

    using BasicPublishPtr = std::shared_ptr<NS_Protocol::BasicPublish>;
    using BasicAckPtr = std::shared_ptr<NS_Protocol::BasicAck>;

    class Channel
    {
    public:
        using Ptr = std::shared_ptr<Channel>;
        Channel(const std::string &channel_id,
                const NS_Host::VirtualHost::Ptr &host,
                const NS_Consumer::ConsumerManager::Ptr &cmp,
                const ProtobufCodecPtr &codec,
                const muduo::net::TcpConnectionPtr &conn,
                const ThreadPoolPtr &thread_pool)
            : _channel_id(channel_id), _host(host), _cmp(cmp), _conn(conn), _codec(codec),
              _thread_pool(thread_pool)
        {
            DLOG("信道创建");
        }

        // 声明/删除交换机
        void declareExchange(const DeclareExchangeRequestPtr &req)
        {
            DLOG("交换机类型：%d", req->type() == NS_Message::ExchangeType::FANOUT);
            bool ret = _host->declareExchange(req->ename(), req->type(), req->durable(), req->auto_delete(), req->args());
            return basicResponse(req->request_id(), req->channel_id(), ret);
        }
        void deleteExchange(const DeleteExchangeRequestPtr &req)
        {
            _host->deleteExchange(req->ename());
            return basicResponse(req->request_id(), req->channel_id(), true);
        }
        // 声明/删除消息队列
        // 在声明和删除消息队列的时候需要创建消费者管理的队列单元
        void declareMsgQueue(const DeclareMsgQueueRequestPtr &req)
        {
            bool ret = _host->declareMsgQueue(req->qname(), req->exclusive(), req->durable(), req->auto_delete(), req->args());
            if (ret == false)
            {
                return basicResponse(req->request_id(), req->channel_id(), ret);
            }
            ret = ret && _cmp->initQueueConsumer(req->qname());
            return basicResponse(req->request_id(), req->channel_id(), ret);
        }
        void deleteMsgQueue(const DeleteMsgQueueRequestPtr &req)
        {
            _cmp->destoryQueueConsumer(req->qname());
            _host->deleteMsgQueue(req->qname());
            return basicResponse(req->request_id(), req->channel_id(), true);
        }
        // 绑定/解绑交换机与消息队列绑定关系
        void Bind(const CreateBindingRequestPtr &req)
        {
            bool ret = _host->declareBinding(req->ename(), req->qname(), req->binding_key());
            return basicResponse(req->request_id(), req->channel_id(), ret);
        }
        void unBind(const DeleteBindingRequestPtr &req)
        {
            _host->deleteBinding(req->ename(), req->qname());
            return basicResponse(req->request_id(), req->channel_id(), true);
        }
        // 消息的发布
        void basicPublic(const BasicPublishPtr &req)
        {
            auto ep = _host->selectExchange(req->ename());
            if (ep == nullptr)
            {
                DLOG("消息发布中，没有找到对应交换机：%s", req->ename().c_str());
                return basicResponse(req->request_id(), req->channel_id(), false);
            }

            NS_Binding::MsgQueueBindingMap mqbm = _host->getExchangeBindings(req->ename());
            NS_Message::MessageProperties *properties = req->mutable_properties();
            std::string routing_key;
            if (properties != nullptr)
            {
                routing_key = properties->routing_key();
            }
            // DLOG("消息是否持久化: %d，绑定信息队列大小: %d，交换机名称：%s, 交换机类型：%d",
            //      properties->mode() == NS_Message::DeliveryMode::DURABLE, (int)mqbm.size(), ep->name.c_str(), ep->type == NS_Message::ExchangeType::FANOUT);
            for (auto [qname, ptr] : mqbm)
            {
                // DLOG("路由匹配：%s, %s", ptr->binding_key.c_str(), routing_key.c_str());
                if (NS_Route::Router::isMatch(ep->type, ptr->binding_key, routing_key))
                {
                    // 3.将消息添加到队列中（添加消息的管理）

                    _host->basicPublish(qname, properties, req->body());
                    // 4.向线程池中添加⼀个消息消费任务(向指定队列的订阅者去推送消息线程池完成)


                    // DLOG("抛入任务: %s, %s, %s", ptr->binding_key.c_str(), routing_key.c_str(), qname.c_str());
                    // _thread_pool->post(std::bind(&Channel::consume, this, "queue1"));
                    // DLOG("虚拟机引用计数：%ld", _host.use_count());
                    static int count = 0;
                    // DLOG("抛入任务，%d", count++);
                    _thread_pool->post(std::bind(&Channel::consume, this, std::move(qname)));
                    // DLOG("抛入任务完成");
                }
            }
            // DLOG("1111111");
            return basicResponse(req->request_id(), req->channel_id(), true);
        }
        // 消息的确认
        void basicAck(const BasicAckPtr &req)
        {
            _host->basicAck(req->qname(), req->msg_id());
            return basicResponse(req->request_id(), req->channel_id(), true);
        }
        // 消息队列的订阅
        void queueSubscription(const CreateConsumerRequestPtr &req)
        {
            // 1. 判断队列是否存在

            bool ret = _host->existsMsgQueue(req->qname());
            if (ret == false)
            {
                DLOG("订阅消息队列时，没有对应的消息队列：%s", req->qname().c_str());
                return basicResponse(req->request_id(), req->channel_id(), ret);
            }
            // 2. 创建队列的消费者

            auto cb = std::bind(&Channel::callback, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3);
            // 创建了消费者之后，当前的channel⻆⾊就是个消费者

            _consumer = _cmp->insert(req->qname(), req->consumer_tag(), req->auto_ack(), cb);
            DLOG("消费者创建成功");
            return basicResponse(req->request_id(), req->channel_id(), true);
        }
        // 消息队列的取消订阅
        void queueCancelSubscription(const DeleteConsumerRequestPtr &req)
        {
            _cmp->erase(req->qname(), req->consumer_tag());
            return basicResponse(req->request_id(), req->channel_id(), true);
        }

        ~Channel()
        {
            // 如果信道与消费者产生关联，那么在信道析构的时候需要删除消费者
            if (_consumer.get() != nullptr)
            {
                _cmp->erase(_consumer->_qname, _consumer->_tag);
            }
            DLOG("信道关闭");
        }

    private:
        void callback(const std::string &qname, const NS_Message::MessageProperties *mp, const std::string &body)
        {
            NS_Protocol::BasicConsumerResponse resp;
            resp.set_channel_id(_channel_id);
            resp.set_consumer_tag(_consumer->_tag);
            if (mp)
            {
                resp.mutable_properties()->set_id(mp->id());
                resp.mutable_properties()->set_mode(mp->mode());
                resp.mutable_properties()->set_routing_key(mp->routing_key());
            }
            resp.set_body(body);
            _codec->send(_conn, resp);
        }

        void consume(const std::string& qname)
        {
            // 指定队列消费消息

            // 1.从队列中取出⼀条消息
            if(_host.get() == nullptr)
            {
                ELOG("程序出现错误，虚拟机指针为空");
                exit(0);
            }

            NS_Message::MessagePtr mp = _host->basicConsume(qname);
            if (mp.get() == nullptr)
            {
                DLOG("执⾏消费任务失败，%s 队列没有消息！", qname.c_str());
                return;
            }
            // 2.从队列订阅者中取出⼀个订阅者
            // DLOG("%s", qname.c_str());
            NS_Consumer::Consumer::Ptr cp = _cmp->choose(qname);
            if (cp.get() == nullptr)
            {
                DLOG("执⾏消费任务失败，%s 队列没有消费者！", qname.c_str());
                return;
            }
            // 3.调⽤订阅者对应的消息处理函数，实现消息的推送
            cp->_callback(cp->_tag, mp->mutable_payload()->mutable_property(), mp->payload().body());
            // 4.判断如果订阅者是⾃动确认-- 不需要等待确认，直接删除消息，

            // 否则需要外部收到消息确认后再删除

            if (cp->_auto_ack)
                _host->basicAck(qname, mp->payload().property().id());
        }

        // 向指定信道发送响应状态
        void basicResponse(const std::string &request_id, const std::string &channel_id, bool ok)
        {
            NS_Protocol::BasicCommonResponse resp;
            resp.set_request_id(request_id);
            resp.set_channel_id(channel_id);
            resp.set_ok(ok);
            _codec->send(_conn, resp);
        }

    private:
        // 信道的id
        std::string _channel_id;
        // 信道所绑定的消费者（如果有的话）
        NS_Consumer::Consumer::Ptr _consumer;
        // 虚拟机管理句柄
        NS_Host::VirtualHost::Ptr _host;
        // 消费者管理句柄
        NS_Consumer::ConsumerManager::Ptr _cmp;
        // TCP连接
        muduo::net::TcpConnectionPtr _conn;
        // muduo库根据protobuf制定的通信报文处理句柄
        ProtobufCodecPtr _codec;
        // 工作线程池句柄
        ThreadPoolPtr _thread_pool;
    };

    class ChannelManager
    {
    public:
        using Ptr = std::shared_ptr<ChannelManager>;
        ChannelManager() {}
        bool openChannel(const std::string &id,
                         const NS_Host::VirtualHost::Ptr &host,
                         const NS_Consumer::ConsumerManager::Ptr &cmp,
                         const ProtobufCodecPtr &codec,
                         const muduo::net::TcpConnectionPtr &conn,
                         const ThreadPoolPtr &pool)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(id);
            if (it != _channels.end())
            {
                DLOG("信道：%s 已经存在!", id.c_str());
                return false;
            }
            auto channel = std::make_shared<Channel>(id, host, cmp, codec, conn, pool);
            _channels.insert(std::make_pair(id, channel));
            return true;
        }
        void closeChannel(const std::string &id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            DLOG("信道%s被关闭", id.c_str());
            _channels.erase(id);
        }
        Channel::Ptr getChannel(const std::string &id)
        {
            std::unique_lock<std::mutex> lock(_mutex);
            auto it = _channels.find(id);
            if (it == _channels.end())
            {
                return Channel::Ptr();
            }
            return it->second;
        }

        ~ChannelManager()
        {
            // for(auto [channel_id, channrl_ptr] : _channels)
            //     closeChannel(channel_id);
        }

    private:
        std::mutex _mutex;
        std::unordered_map<std::string, Channel::Ptr> _channels;
    };

}

#endif